package org.tech4j.springboot.starter.logging.kafka;

import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.tech4j.logging.common.client.LogConsumeClient;
import org.tech4j.logging.common.collector.LogCollector;
import org.tech4j.logging.common.config.LogThreadPoolConfig;
import org.tech4j.logging.common.utils.LogCollectConfigUtils;
import org.tech4j.logging.kafka.client.KafkaLogCollectClient;
import org.tech4j.logging.kafka.collector.KafkaLogCollector;
import org.tech4j.logging.kafka.config.KafkaLogCollectConfig;

/**
 * @program: tech4j
 * @description:
 * @author: YuKai Fan
 * @create: 2025/3/5 21:23
 **/
@Configuration
@ConditionalOnProperty(value = {"tech4j.logging-kafka.enabled"}, havingValue = "true", matchIfMissing = false)
public class LoggingKafkaConfiguration {

    @Bean
    @ConfigurationProperties("tech4j.logging-kafka")
    public KafkaLogCollectConfig kafkaLogCollectConfig() {
        return new KafkaLogCollectConfig();
    }

    @Bean
    @ConfigurationProperties("tech4j.logging-kafka.thread-pool")
    public LogThreadPoolConfig logThreadPoolConfig() {
        LogThreadPoolConfig logThreadPoolConfig = new LogThreadPoolConfig();
        LogCollectConfigUtils.setLogThreadPoolConfig(logThreadPoolConfig);
        return logThreadPoolConfig;
    }

    @Bean
    @ConditionalOnMissingBean(LogConsumeClient.class)
    public KafkaLogCollectClient kafkaLogCollectClient(final KafkaLogCollectConfig kafkaLogCollectConfig) {
        try (KafkaLogCollectClient kafkaLogCollectClient = new KafkaLogCollectClient(kafkaLogCollectConfig)){
            kafkaLogCollectClient.initClient(kafkaLogCollectConfig.getKafkaLogConfig());
            return kafkaLogCollectClient;
        }
    }

    @Bean
    @ConditionalOnMissingBean(LogCollector.class)
    public KafkaLogCollector kafkaLogCollector(final KafkaLogCollectClient kafkaLogCollectClient) {
        KafkaLogCollector kafkaLogCollector = new KafkaLogCollector(kafkaLogCollectClient);
        kafkaLogCollector.start();
        return kafkaLogCollector;
    }
}
